package com.shujia.car.batch.jtqj

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object DwsAppVioZqzf {
  def main(args: Array[String]): Unit = {

    val ds: String = args.head

    val spark: SparkSession = SparkSession
      .builder()
      .appName("DwsAppVioZqzf")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    //现场违法表
    val baseVioForce: DataFrame = spark
      .table("dwd.dwd_base_vio_force")
      .where($"ds" === ds)

    //非现场违法表
    val baseVioSurveil: DataFrame = spark
      .table("dwd.dwd_base_vio_surveil")
      .where($"ds" === ds)

    //提取现场违法表中的违法行为
    val baseVioForceWfxw: DataFrame = baseVioForce
      .select(
        $"xh",
        date_format($"wfsj", "yyyy-MM-dd") as "wfsj",
        explode(array($"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")) as "wfxw" //一行转多行
      )
      .withColumn("wfxw", regexp_replace($"wfxw", "[A-Z]$", ""))
      .where($"wfxw" =!= "NULL")

    //提取非现场违法表中的违法行为
    val baseVioSurveilWfxw: DataFrame = baseVioSurveil
      .select(
        $"xh",
        date_format($"wfsj", "yyyy-MM-dd") as "wfsj",
        regexp_replace($"wfxw", "[A-Z]$", "") as "wfxw"
      )

    //合并现场违法表和非现场违法表
    val unionWfxw: DataFrame = baseVioForceWfxw.union(baseVioSurveilWfxw)

    //获取指标口径
    val kjMap: Map[String, List[String]] = ZqzfKjUtil.getKJ

    //统计指标
    var baseIndex: DataFrame = unionWfxw
      .groupBy($"wfsj")
      .agg(
        countDistinct($"xh") as "dr_zd_wfs", //当日总的违法数
        sum(when($"wfxw".isInCollection(kjMap("zjjs")), 1).otherwise(0)) as "dr_zjjs_wfs",
        sum(when($"wfxw".isInCollection(kjMap("yjjs")), 1).otherwise(0)) as "dr_yjjs_wfs",
        sum(when($"wfxw".isInCollection(kjMap("zdhp")), 1).otherwise(0)) as "dr_zdhp_wfs",
        sum(when($"wfxw".isInCollection(kjMap("wzbz")), 1).otherwise(0)) as "dr_wzbz_wfs",
        sum(when($"wfxw".isInCollection(kjMap("tpjp")), 1).otherwise(0)) as "dr_tpjp_wfs",
        sum(when($"wfxw".isInCollection(kjMap("wdtk")), 1).otherwise(0)) as "dr_wdtk_wfs",
        sum(when($"wfxw".isInCollection(kjMap("jspz")), 1).otherwise(0)) as "dr_jspz_wfs",
        sum(when($"wfxw".isInCollection(kjMap("wxjs")), 1).otherwise(0)) as "dr_wxjs_wfs",
        sum(when($"wfxw".isInCollection(kjMap("cy")), 1).otherwise(0)) as "dr_cy_wfs",
        sum(when($"wfxw".isInCollection(kjMap("cz")), 1).otherwise(0)) as "dr_cz_wfs"
      )

    //指标列表
    val indexs: List[String] = List("zd", "zjjs", "yjjs", "zdhp", "wzbz", "tpjp", "wdtk", "jspz", "wxjs", "cy", "cz")

    //将一样的代码逻辑放到循环中
    for (index <- indexs) {
      //每一次计算基于上一次的结果计算
      baseIndex = baseIndex
        //今年违法数
        .withColumn(s"jn_${index}_wfs", sum($"dr_${index}_wfs") over Window.partitionBy(date_format($"wfsj", "yyyy")))
        //上月当日违法数
        .withColumn(s"sy_dr_${index}_wfs", lag($"dr_${index}_wfs", 1) over Window.partitionBy(date_format($"wfsj", "dd")).orderBy(date_format($"wfsj", "yyyy-MM")))
        //计算同比
        .withColumn(s"${index}_tb", round(coalesce($"dr_${index}_wfs" / $"sy_dr_${index}_wfs" - 1, expr("1.0")), 4))
    }

    //整理数据
    val resultDF: DataFrame = baseIndex.select(
      $"wfsj",
      $"dr_zd_wfs",
      $"jn_zd_wfs",
      $"zd_tb",
      $"dr_zjjs_wfs",
      $"jn_zjjs_wfs",
      $"zjjs_tb",
      $"dr_yjjs_wfs",
      $"jn_yjjs_wfs",
      $"yjjs_tb",
      $"dr_zdhp_wfs",
      $"jn_zdhp_wfs",
      $"zdhp_tb",
      $"dr_wzbz_wfs",
      $"jn_wzbz_wfs",
      $"wzbz_tb",
      $"dr_tpjp_wfs",
      $"jn_tpjp_wfs",
      $"tpjp_tb",
      $"dr_wdtk_wfs",
      $"jn_wdtk_wfs",
      $"wdtk_tb",
      $"dr_jspz_wfs",
      $"jn_jspz_wfs",
      $"jspz_tb",
      $"dr_wxjs_wfs",
      $"jn_wxjs_wfs",
      $"wxjs_tb",
      $"dr_cy_wfs",
      $"jn_cy_wfs",
      $"cy_tb",
      $"dr_cz_wfs",
      $"jn_cz_wfs",
      $"cz_tb"
    )

    //保存数据
    resultDF.createOrReplaceTempView("tmp")
    spark.sql(
      s"""
         |insert overwrite table dws.dws_app_vio_zqzf partition(ds='${ds}')
         |select * from tmp
         |""".stripMargin)


  }

}
